package com.yeyks.config;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.aliyun.openservices.ons.api.batch.BatchConsumer;
import com.yeyks.common.constants.rocketmq.TopicAndTag;
import com.yeyks.rocketmq.MessageCnBookingOrderListener;
import com.yeyks.rocketmq.MessageCnBookingPushListener;
import com.yeyks.rocketmq.MessageCnBookingSearchListener;
import com.yeyks.rocketmq.MessageESUpdateListener;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Properties;


@Data
@Slf4j
@Component
public class RocketMqConsumer {
    @Autowired
    private RocketMqConfig rocketMqConfig;

    @Bean(name = "esUpdateConsumer")
    public Consumer getCToWebOrderConsumer(MessageESUpdateListener messageESUpdateListener) {
        log.info("初始化ES房态启动消费者！");
        // producer 实例配置初始化
        Properties properties = rocketMqConfig.getProperties();
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSDELAY_ESUPDATE);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_ESUPDATE, messageESUpdateListener);
        consumer.start();
        log.info("启动更新ES中房态数据消费者成功！");
        return consumer;
    }

    /**
     * 延迟更新订单状态
     *
     * @return
     */
    @Bean(name = "CNBookingOrderConsumer")
    public Consumer getCNBookingOrderConsumer(MessageCnBookingOrderListener cnBookingOrderListener) {
        log.info("初始化延迟启动消费者！");
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_YYKSNORMAL_CNBACK);
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_CNBOOKING_ORDER_UPATE, cnBookingOrderListener);
        consumer.start();
        log.info("启动延迟消费者成功！");
        return consumer;
    }

    /**
     * CN 推送消费
     *
     * @return 返回消费信息
     */
    @Bean(name = "CNBookingPushConsumer")
    public Consumer getCNBookingPushConsumer(MessageCnBookingPushListener cnBookingPushListener) {
        log.info("初始化CN 推送消费");
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_CN_YYKSNORMAL_CNPUSH);
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(TopicAndTag.TOPIC_YYKS_DELAY, TopicAndTag.TAG_CN_CNBOOKING_PUSH, cnBookingPushListener);
        consumer.start();
        log.info("启动CN 推送消费成功！");
        return consumer;
    }

    /**
     * 消费搜索发送的消息
     * @param cnBookingSearchListener 监听
     * @return 响应消息
     */
    @Bean(name = "SearchPushConsumer")
    public BatchConsumer getSearchPushConsumer(MessageCnBookingSearchListener cnBookingSearchListener) {
        log.info("初始化搜索 推送消费");
        Properties properties = rocketMqConfig.getProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, TopicAndTag.GID_WEB_TO_B_FOR_CN_HOTEL_SYNC);
        // 集群订阅方式
        properties.setProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        properties.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, "10");
        BatchConsumer batchConsumer = ONSFactory.createBatchConsumer( properties );
        batchConsumer.subscribe(TopicAndTag.TOPIC_YYKS_NORMAL, TopicAndTag.TAG_WEB_TO_B_FOR_CN_HOTEL_SYNC, cnBookingSearchListener);
        batchConsumer.start();
        log.info("启动搜索 推送消费成功！");
        return batchConsumer;
    }


}
